package com.burges.net.assembly.cep

import org.apache.flink.api.common.time.Time
import org.apache.flink.cep.CEP
import org.apache.flink.cep.pattern.Pattern
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}

/**
  * 创建人    BurgessLee 
  * 创建时间   2020/2/26 
  * 描述     Flink CEP应用代码示例
  */
object FlinkCEPDEMO {

	def main(args: Array[String]): Unit = {
		val env = StreamExecutionEnvironment.getExecutionEnvironment
		//设定时间概念属性
		env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
		//创建输入事件流
		val input: DataStream[Event] = env.fromElements(...)
		//根据ID分区
		val partitionedInput = input.keyBy(event => event.getId)
		//创建Pattern
		val pattern = Pattern.begin[Event]("start")
        		.next("middle")
        		.where((event, ctx) => event.getType == "temperature")
        		.followedBy("end")
        		.where((event,ctx) => event.getId >= 1000)
		//指定超时限制
        		.within(Time.seconds(10))
		//通过CEP方法创建PatternStream
		val patternStream = CEP.pattern(input, pattern)
		//调用select方法
		val result : DataStream[Event] = patternStream.select(event => selectFn(event))
	}

	def selectFn(pattern: Map[String, Iterable[Event]]): Event = {
		//获取pattern中的startEvent
		val startEvent = pattern.get("start").iterator.next().toList(0)
		startEvent
	}

}
